package cn.ymatrix;

import cn.ymatrix.connectors.MySqlToYMatrix;
import cn.ymatrix.service.ConnectService;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MySQLSinkYmatrix {

    public static void main(String[] args) throws Exception {

        // MYSQL 案例
        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname(ConnectService.MYSQL_IP)
                .port(ConnectService.MYSQL_PORT)
                .databaseList(ConnectService.MYSQL_DB)
                .username(ConnectService.MYSQL_USERNAME)
                .password(ConnectService.MYSQL_PASSWORD)
                .deserializer(new MySqlToYMatrix())
                // .deserializer(new StringDebeziumDeserializationSchema())
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // env.enableCheckpointing(3000);
        env.addSource(sourceFunction).print().setParallelism(3).setParallelism(10);
        env.execute("Print MySQL Snapshot + Binlog");
    }
}